%% copy riak-0.14.2 source to learn dynamo implementation and erlang
%% author: lichuang
% copy source from riak_kv_put_fsm.erl
-module(edynamo_put_fsm).
-include_lib("riak_kv_vnode.hrl").
-include("riakc_obj.hrl").
%-include("riak_kv_wm_raw.hrl").
-behaviour(gen_fsm).
-define(DEFAULT_OPTS, [{returnbody, false}, {update_last_modified, true}]).
-export([start/6,start/7]).
-export([init/1, handle_event/3, handle_sync_event/4,
         handle_info/3, terminate/3, code_change/4]).
-export([initialize/2, waiting_vnode_w/2, waiting_vnode_dw/2]).

-record(state, {robj :: edynamo_object:edynamo_object(),
                client :: {pid(), reference()},
                rclient :: edynamo_client:edynamo_client(),
                n :: pos_integer(),
                w :: pos_integer(),
                dw :: non_neg_integer(),
                preflist :: [{pos_integer(), atom()}],
                bkey :: {edynamo_object:bucket(), edynamo_object:key()},
                waiting_for :: list(),
                req_id :: pos_integer(),
                starttime :: pos_integer(),
                replied_w :: list(),
                replied_dw :: list(),
                replied_fail :: list(),
                timeout :: pos_integer(),
                tref    :: reference(),
                ring :: edynamo_ring:edynamo_ring(),
                startnow :: {pos_integer(), pos_integer(), pos_integer()},
                vnode_options :: list(),
                returnbody :: boolean(),
                resobjs :: list(),
                allowmult :: boolean(),
                update_last_modified :: boolean()
               }).

start(ReqId, RObj, W, DW, Timeout, From) ->
  start(ReqId, RObj, W, DW, Timeout, From, []).

start(ReqId, RObj, W, DW, Timeout, From, Options) ->
  gen_fsm:start(?MODULE, [ReqId, RObj, W, DW, Timeout, From, Options], []).

init([ReqId, RObj0, W0, DW0, Timeout, Client, Options0]) ->
  Options = flatten_options(proplists:unfold(Options0 ++ ?DEFAULT_OPTS), []),
  {ok, Ring} = edynamo_ring_manager:get_my_ring(),
  BucketProps = edynamo_bucket:get_bucket(edynamo_object:bucket(RObj0), Ring),
  N = proplists:get_value(n_val, BucketProps),
  W = edynamo_util:expand_rw_value(w, W0, BucketProps, N),

  %% Expand the DW value, but also ensure that DW <= W
  DW = erlang:min(edynamo_util:expand_rw_value(dw, DW0, BucketProps, N), W),

  case (W > N) or (DW > N) of
    true ->
      Client ! {ReqId, {error, {n_val_violation, N}}},
      {stop, normal, none};
    false ->
      AllowMult = proplists:get_value(allow_mult, BucketProps),
      {ok, RClient} = edynamo_util:local_client(),
      Bucket = edynamo_object:bucket(RObj0),
      Key = edynamo_object:key(RObj0),
      StateData0 = #state{robj = RObj0,
			  client = Client, w = W, dw = DW, bkey = {Bucket, Key},
			  req_id = ReqId, timeout = Timeout, ring = Ring,
			  rclient = RClient,
			  vnode_options = [],
			  resobjs = [], allowmult = AllowMult},
      StateData = handle_options(Options, StateData0),
      {ok, initialize, StateData, 0}
  end.

handle_event(_Event, _StateName, StateData) ->
  {stop,badmsg,StateData}.

handle_sync_event(_Event, _From, _StateName, StateData) ->
  {stop,badmsg,StateData}.

handle_info(timeout, StateName, StateData) ->
  ?MODULE:StateName(timeout, StateData);
handle_info(_Info, _StateName, StateData) ->
  {stop,badmsg,StateData}.

terminate(Reason, _StateName, _State) ->
  Reason.

code_change(_OldVsn, StateName, State, _Extra) ->
  {ok, StateName, State}.

%%
%% Given an expanded proplist of options, take the first entry for any given key
%% and ignore the rest
%%
%% @private
flatten_options([], Opts) ->
  Opts;
flatten_options([{Key, Value} | Rest], Opts) ->
  case lists:keymember(Key, 1, Opts) of
    true ->
      flatten_options(Rest, Opts);
    false ->
      flatten_options(Rest, [{Key, Value} | Opts])
  end.

%% @private
handle_options([], State) ->
  State;
handle_options([{update_last_modified, Value} | T], State) ->
  handle_options(T, State#state{update_last_modified=Value});
handle_options([{returnbody, true} | T], State) ->
  VnodeOpts = [{returnbody, true} | State#state.vnode_options],
  %% Force DW>0 if requesting return body to ensure the dw event 
  %% returned by the vnode includes the object.
  handle_options(T, State#state{vnode_options=VnodeOpts,
				dw=erlang:max(1, State#state.dw),
				returnbody=true});
handle_options([{returnbody, false} | T], State) ->
  case has_postcommit_hooks(element(1, State#state.bkey)) of
    true ->
      %% We have post-commit hooks, we'll need to get the body back
      %% from the vnode, even though we don't plan to return that to the
      %% original caller.  Force DW>0 to ensure the dw event returned by
      %% the vnode includes the object.
      VnodeOpts = [{returnbody, true} | State#state.vnode_options],
      handle_options(T, State#state{vnode_options=VnodeOpts,
				    dw = erlang:max(1, State#state.dw),
				    returnbody=false});
    false ->
      handle_options(T, State#state{returnbody=false})
  end;
handle_options([{_,_} | T], State) ->
  handle_options(T, State).

initialize(timeout, StateData0=#state{robj=RObj0, req_id = ReqId, client = Client,
                                      update_last_modified = UpdateLastMod,
                                      timeout = Timeout, ring = Ring, bkey = {Bucket, Key} = BKey,
                                      rclient = RClient, vnode_options = VnodeOptions}) ->
  case invoke_hook(precommit, RClient, update_last_modified(UpdateLastMod, RObj0)) of
    fail ->
      Client ! {ReqId, {error, precommit_fail}},
      {stop, normal, StateData0};
    {fail, Reason} ->
      Client ! {ReqId, {error, {precommit_fail, Reason}}},
      {stop, normal, StateData0};
    RObj1 ->
      StartNow = now(),
      TRef = erlang:send_after(Timeout, self(), timeout),
      RealStartTime = edynamo_util:moment(),
      BucketProps = edynamo_bucket:get_bucket(Bucket, Ring),
      DocIdx = edynamo_util:chash_key({Bucket, Key}),
      Req = ?KV_PUT_REQ{
        bkey = BKey,
        object = RObj1,
        req_id = ReqId,
        start_time = RealStartTime,
        options = VnodeOptions},
      N = proplists:get_value(n_val, BucketProps),
      Preflist = edynamo_ring:preflist(DocIdx, Ring),
      error_logger:error_msg("preflist:~p~n", [Preflist]),
      {Targets, Fallbacks} = lists:split(N, Preflist),
      UpNodes = edynamo_node_watcher:nodes(edynamo),
      {Sent1, Pangs1} = edynamo_util:try_cast(Req, UpNodes, Targets),
      Sent = case length(Sent1) =:= N of %  Sent is [{Index,TargetNode,SentNode}]
              true  -> Sent1;
        false -> Sent1 ++ edynamo_util:fallback(Req, UpNodes, Pangs1, Fallbacks)
             end,
      StateData = StateData0#state{
                    robj = RObj1, n = N, preflist = Preflist,
                    waiting_for = Sent, starttime = edynamo_util:moment(),
                    replied_w = [], replied_dw = [], replied_fail = [],
                    tref = TRef, startnow = StartNow},
      {next_state, waiting_vnode_w, StateData}
  end.
waiting_vnode_w({w, Idx, ReqId},
                StateData=#state{w = W, dw = DW, req_id = ReqId,
                client = Client, replied_w = Replied0}) ->
  Replied = [Idx | Replied0],
  case length(Replied) >= W of
    true ->
      case DW of
        0 ->
          Client ! {ReqId, ok},
          %update_stats
          {stop, normal, StateData};
        _ ->
          NewStateData = StateData#state{replied_w = Replied},
          {next_state, waiting_vnode_dw, NewStateData}
      end;
    false ->
      NewStateData = StateData#state{replied_w = Replied},
      {next_state, waiting_vnode_w, NewStateData}
  end;
waiting_vnode_w({dw, Idx, _ReqId},
                StateData = #state{replied_dw = Replied0}) ->
  Replied = [Idx | Replied0],
  NewStateData = StateData#state{replied_dw = Replied},
  {next_state, waiting_vnode_w, NewStateData};
waiting_vnode_w({dw, Idx, ResObj, _ReqId},
                StateData=#state{replied_dw = Replied0, resobjs = ResObjs0}) ->
  Replied = [Idx | Replied0],
  ResObjs = [ResObj | ResObjs0],
  NewStateData = StateData#state{replied_dw = Replied, resobjs = ResObjs},
  {next_state, waiting_vnode_w, NewStateData};
waiting_vnode_w({fail, Idx, ReqId},
                StateData = #state{n = N, w = W, client = Client,
                                   replied_fail = Replied0}) ->
  Replied = [Idx | Replied0],
  NewStateData = StateData#state{replied_fail = Replied},
  case (N - length(Replied)) >= W of
    true ->
      {next_state, waiting_vnode_w, NewStateData};
    false ->
      %% update_stats
      Client ! {ReqId, {error, too_many_fails}},
      {stop, normal, NewStateData}
  end;
waiting_vnode_w(timeout, StateData=#state{client = Client, req_id = ReqId}) ->
  %% update_stats
  Client ! {ReqId, {error, timeout}},
  {stop, normal, StateData}.

waiting_vnode_dw({w, _Idx, ReqId},
                 StateData=#state{req_id=ReqId}) ->
  {next_state, waiting_vnode_dw, StateData};
waiting_vnode_dw({dw, Idx, ReqId},
                 StateData=#state{dw=DW, client=Client,replied_dw=Replied0}) ->
  Replied = [Idx | Replied0],
  case length(Replied) >= DW of
    true ->
      Client ! {ReqId, ok},
      %%update_stats
      {stop, normal, StateData};
    false ->
      NewStateData = StateData#state{replied_dw=Replied},
      {next_state, waiting_vnode_dw, NewStateData}
  end;
waiting_vnode_dw({dw, Idx, ResObj, ReqId},
                 StateData=#state{dw=DW, client=Client,replied_dw = Replied0,
                                  allowmult=AllowMult, returnbody=ReturnBody,
                                  rclient=RClient, resobjs=ResObjs0}) ->
  Replied = [Idx | Replied0],
  ResObjs = [ResObj | ResObjs0],
  case length(Replied) >= DW of
    true ->
      ReplyObj = merge_robjs(ResObjs, AllowMult),
      Reply = case ReturnBody of
                true  -> {ok, ReplyObj};
                false -> ok
              end,
      Client ! {ReqId, Reply},
      invoke_hook(postcommit, RClient, ReplyObj),
      %% update_stats
      {stop, normal, StateData};
    false ->
      NewStateData = StateData#state{replied_dw=Replied, resobjs = ResObjs},
      {next_state, waiting_vnode_dw, NewStateData}
  end;
waiting_vnode_dw({fail, Idx, ReqId},
                  StateData = #state{n = N, dw = DW, client = Client,
                                     replied_fail = Replied0}) ->
  Replied = [Idx | Replied0],
  NewStateData = StateData#state{replied_fail = Replied},
  case (N - length(Replied)) >= DW of
    true ->
      {next_state, waiting_vnode_dw, NewStateData};
    false ->
      Client ! {ReqId, {error, too_many_fails}},
      {stop, normal, NewStateData}
  end;
waiting_vnode_dw(timeout, StateData=#state{client = Client, req_id = ReqId}) ->
  %% update_stats
  Client ! {ReqId, {error, timeout}},
  {stop, normal, StateData}.

%% Internal functions
invoke_hook(HookType, RClient, RObj) ->
  Bucket = edynamo_object:bucket(RObj),
  BucketProps = RClient:get_bucket(Bucket),
  R = proplists:get_value(HookType, BucketProps, []),
  case R of
    <<"none">> ->
      RObj;
    [] ->
      RObj;
    Hooks when is_list(Hooks) ->
      run_hooks(HookType, RObj, Hooks)
  end.      
invoke_hook(precommit, Mod0, Fun0, undefined, RObj) ->
  Mod = binary_to_atom(Mod0, utf8),
  Fun = binary_to_atom(Fun0, utf8),
  wrap_hook(Mod, Fun, RObj);
invoke_hook(precommit, undefined, undefined, _JSName, _RObj) ->
  error_logger:warning_msg("Javascript post-commit hooks aren't implemented");
invoke_hook(postcommit, Mod0, Fun0, undefined, Obj) ->
  Mod = binary_to_atom(Mod0, utf8),
  Fun = binary_to_atom(Fun0, utf8),
  proc_lib:spawn(fun() -> wrap_hook(Mod, Fun, Obj) end);
invoke_hook(postcommit, undefined, undefined, _JSName, _Obj) ->
  error_logger:warning_msg("Javascript post-commit hooks aren't implemented");
%% NOP to handle all other cases
invoke_hook(_, _, _, _, RObj) ->
  RObj.

wrap_hook(Mod, Fun, Obj) ->
  try Mod:Fun(Obj)
  catch 
    EType:X ->
      error_logger:error_msg("problem invoking hook ~p:~p -> ~p:~p~n~p~n",
        [Mod,Fun,EType,X,erlang:get_stacktrace()]),
      fail
  end.

run_hooks(_HookType, RObj, []) ->
  RObj;
run_hooks(HookType, RObj, [{struct, Hook} | T]) ->
  Mod = proplists:get_value(<<"mod">>, Hook),
  Fun = proplists:get_value(<<"fun">>, Hook),
  JSName = proplists:get_value(<<"name">>, Hook),
  Result = invoke_hook(HookType, Mod, Fun, JSName, RObj),
  case HookType of
    precommit ->
      case Result of
        fail ->
          Result;
        _ ->
          run_hooks(HookType, Result, T)
      end;
    postcommit ->
      run_hooks(HookType, RObj, T)
  end.

%%
%% Update X-Riak-VTag and X-Riak-Last-Modified in the object's metadata, if
%% necessary.
%%
%% @private
update_last_modified(false, RObj) ->
  RObj;
update_last_modified(true, RObj)  ->
  MD0 = case dict:find(clean, edynamo_object:get_update_metadata(RObj)) of
          {ok, true} ->
            %% There have been no changes to updatemetadata. If we stash the
            %% last modified in this dict, it will cause us to lose existing
            %% metadata (bz://508). If there is only one instance of metadata,
            %% we can safely update that one, but in the case of multiple siblings,
            %% it's hard to know which one to use. In that situation, use the update
            %% metadata as is.
            case edynamo_object:get_metadatas(RObj) of
              [MD] ->
                MD;
              _ ->
                edynamo_object:get_update_metadata(RObj)
            end;
          _ ->
            edynamo_object:get_update_metadata(RObj)
      end,
  error_logger:error_msg("md:~p~n", [MD0]),
  NewMD = dict:store(?MD_VTAG, make_vtag(RObj),
                     dict:store(?MD_LASTMOD, erlang:now(), MD0)),    
  edynamo_object:apply_updates(edynamo_object:update_metadata(RObj, NewMD)).

make_vtag(RObj) ->
  error_logger:error_msg("vclock2222:~p~n", [edynamo_object:vclock(RObj)]),
  %error_logger:error_msg("binary:~p~n", [term_to_binary(edynamo_object:vclock(RObj))]),
  <<HashAsNum:128/integer>> = crypto:md5(term_to_binary(edynamo_object:vclock(RObj))),
  edynamo_util:integer_to_list(HashAsNum, 62).

merge_robjs(RObjs0, AllowMult) ->
  RObjs1 = [X || X <- RObjs0,
                 X /= undefined],
  case RObjs1 of
    [] -> {error, notfound};
    _  -> edynamo_object:reconcile(RObjs1, AllowMult)
  end.

has_postcommit_hooks(Bucket) ->
  lists:flatten(
    proplists:get_all_values(
      postcommit,
      edynamo_bucket:get_bucket(Bucket))) /= [].
